package org.example.window;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/19 15:53
 */
public class WindowFullFunctionsApiDemo01 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = environment.socketTextStream("127.0.0.1", 7777);

        KeyedStream<WaterSensor, String> keyedStream = streamSource.map(new WaterSensorFunction())
                .keyBy(WaterSensor::getId);

        keyedStream
                // 时间窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<WaterSensor, String, String, TimeWindow>.Context context,
                                        Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        long end = context.window().getEnd();
                        long start = context.window().getStart();
//
                        String startTime = DateFormatUtils.format(start, "yyyy-MM-dd HH:mm:ss");
                        String endTime = DateFormatUtils.format(end, "yyyy-MM-dd HH:mm:ss");
                        out.collect("窗口开始时间：" + startTime +
                                " 窗口结束时间：" + endTime
                                + " 窗口内数据条数：" + elements.spliterator().estimateSize()
                                + " 窗口内数据：" + elements);

                    }
                })
                .print()
        ;

        environment.execute();
    }

}
